package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTestExam {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 设置状态的失效时间
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL("dwd_test_exam_group"));

        //过滤出试卷课程数据
        Table testPaper = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['course_id'] course_id\n" +
                "from `topic_db` where `table` = 'test_paper' " +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("test_paper", testPaper);

        //过滤出测验数据
        Table testExam = tableEnv.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['paper_id'] paper_id,\n" +
                "data['score'] score,\n" +
                "data['duration_sec'] duration_sec,\n" +
                "ts \n" +
                "from `topic_db`\n" +
                "where `table` = 'test_exam'\n" +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("test_exam", testExam);

        //将上述2张表进行关联
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "te.id,\n" +
                "te.user_id,\n" +
                "te.paper_id,\n" +
                "te.score,\n" +
                "tp.course_id,\n" +
                "te.duration_sec,\n" +
                "te.ts \n" +
                "from test_exam te \n" +
                "join test_paper tp\n" +
                "on te.paper_id = tp.id\n");
        tableEnv.createTemporaryView("result_table", resultTable);


        //将关联的结果写到kafka主题
        //9.1 创建动态表和要写入的kafka主题进行映射
        tableEnv.executeSql("create table dwd_test_exam(\n" +
                "id string,\n" +
                "user_id string,\n" +
                "paper_id string,\n" +
                "score string,\n" +
                "course_id string,\n" +
                "duration_sec string,\n" +
                "ts string,\n" +
                "primary key(id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_test_exam"));

        //9.2 写入
        tableEnv.executeSql("insert into dwd_test_exam select * from result_table");

    }
}
